//
// Created by 郝羽 on 2018/6/12.
//
#include "subprocess.hpp"

#include <glib-2.0/glib.h>
#include <glib-2.0/gio/gio.h>

#include <rapidjson/document.h>
#include <rapidjson/reader.h>
#include <rapidjson/stringbuffer.h>

#include "xm_log.h"

struct subprocess {
    int id;
    GSubprocess* proc;
    subprocess_wait_callback wait_callback;
    void* callback_data;
};

void finish_wait(GObject *source_object, GAsyncResult *res, gpointer user_data) {
    g_message("finish wait");
    auto sub = static_cast<subprocess*>(user_data);
    GError* err = nullptr;
    auto ret = g_subprocess_wait_finish(sub->proc, res, &err);
    if (err != nullptr) {
        g_warning("subprocess: %d reap error: %s", sub->id, err->message);
        g_subprocess_force_exit(sub->proc);
    }
    g_message("g_subprocess_wait_finish ret %s", ret ? "yes" : "no");
    sub->wait_callback(sub, sub->callback_data);
}

subprocess* subprocess_spawn(int id, uint16_t metric_port, Context &ctx,
         subprocess_wait_callback wait_cb, void* data) {
    subprocess* sub = (subprocess*)g_malloc(sizeof(subprocess));
    sub->id = id;
    sub->callback_data = data;

    const gchar* args[1024];
    int n = 0;

    int port_count = (ctx.maxMediaPort - ctx.minMediaPort + 1) / ctx.workers;
    int port_start = ctx.minMediaPort + port_count * id;
    int port_stop = port_start + port_count - 1;

    args[n++] = "bench";
    if (ctx.logConfigFile != nullptr) {
        args[n++] = "--log-config";
        args[n++] = ctx.logConfigFile;
    }
    args[n++] = "--rtc-server";
    args[n++] = ctx.rtcAddresses[id % ctx.rtcAddresses.size()].c_str();
    args[n++] = "--worker";
    args[n++] = "1";
    args[n++] = "--concurrency";
    args[n++] = g_strdup_printf("%d", ctx.concurrency);
    args[n++] = "--ramp-up";
    args[n++] = g_strdup_printf("%d", ctx.rampUpSeconds);
    args[n++] = "--duration";
    args[n++] = g_strdup_printf("%d", ctx.durationSeconds);
    args[n++] = "--media-address";
    args[n++] = ctx.mediaAddress.c_str();
    args[n++] = "--child";
    args[n++] = "--metric-server";
    args[n++] = g_strdup_printf("127.0.0.1:%d", metric_port);
    args[n++] = "--min-port";
    args[n++] = g_strdup_printf("%d", port_start);
    args[n++] = "--max-port";
    args[n++] = g_strdup_printf("%d", port_stop);
    args[n++] = "--video-bitrate";
    args[n++] = g_strdup_printf("%d", ctx.video_bitrate);
//    if (ctx.n_pub != 0) {
//        args[n++] = "--pubs";
//        args[n++] = g_strdup_printf("%d", ctx.n_pub);
//    }
    if (ctx.mixer_id != nullptr) {
        args[n++] = "--mixer-id";
        args[n++] = ctx.mixer_id;
    }
    if (id == 0) {
        for (auto fn : ctx.audio_files) {
            args[n++] = "--audio";
            args[n++] = fn;
        }
    }

    for (auto rtc_id : ctx.pub_rtc_ids) {
        args[n++] = "--pub-rtc";
        args[n++] = rtc_id;
    }

    args[n++] = nullptr;

    std::ostringstream cmd;
    for (auto arg : args) {
        if (arg == nullptr) {
            break;
        }
        cmd << std::string(arg) << " ";
    }
    std::string cmd_str = cmd.str();
    g_message("starting child %d: %s", id, cmd_str.c_str());

    GError* err = nullptr;

    sub->proc = g_subprocess_newv(args,
            GSubprocessFlags(G_SUBPROCESS_FLAGS_INHERIT_FDS),
            &err);

    if (err != nullptr) {
        g_warning("subprocess: %d spawn error: %s", id, err->message);
        exit(-1);
    }

    sub->wait_callback = wait_cb;

    g_subprocess_wait_async(sub->proc, nullptr, finish_wait, sub);

    return sub;
}

GInputStream *subprocess_get_stderr(subprocess *sub) {
    return g_subprocess_get_stderr_pipe(sub->proc);
}

void subprocess_signal(subprocess* sub, gint sig) {
    g_subprocess_send_signal(sub->proc, sig);
}


void subprocess_free(subprocess *sub) {
    g_free(sub);
}



